package org.liang.code.rocketMQ;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.nio.charset.StandardCharsets;

@Slf4j
public class 发送消息 {

    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {

        // 创建DefaultMQProducer消息生产者对象
        DefaultMQProducer producer = new DefaultMQProducer("pro1");

        // 设置NameServer
        producer.setNamesrvAddr("127.0.0.1:9876");

        // 在异步模式下声明发送失败之前内部执行的最大重试次数
        producer.setRetryTimesWhenSendAsyncFailed(3);

        // 与NameServer建立长连接
        producer.start();

        for (int i = 0; i < 5; i++) {

            String data = String.valueOf(i);

             /*创建消息
                Message消息三个参数
                topic 代表消息主题
                tags 代表标志，用于消费者接收数据时进行数据筛选。
                keys
                body 代表消息内容
            */
            Message message = new Message("TTmusic", "", "1", data.getBytes(StandardCharsets.UTF_8));

            // 设置延迟时间秒数
            message.setDelayTimeSec(3);

            // 发送消息
            SendResult result = producer.send(message, (mqs, msg, arg) -> {

                log.info("当前队列数量：" + mqs.size() + ",明细：" + mqs);
                log.info("Message对象：" + msg.toString());
                int dataId = Integer.parseInt(msg.getKeys());
                int index = dataId % mqs.size();
                MessageQueue messageQueue = mqs.get(index);
                log.info("分区队列:" + messageQueue);
                return messageQueue;
            }, null);

            // 将发送结果对象打印在控制台
            System.out.println("消息已发送：MsgId:" + result.getMsgId() + "，发送状态:" + result.getSendStatus());
        }

        // 关闭生产者
        producer.shutdown();

    }
}
